package cn.demo.rocketmq.transaction.demo1;


import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RmqProducerConfig {

    /**
     * 所属的组名
     */
    @Value("${rocketmq.producerGroup}")
    private String producerGroup;

    /**
     * nameserver地址
     */
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;


    @Bean(name = "transactionMQProducer", destroyMethod = "shutdown", initMethod = "start")
    public TransactionMQProducer transactionMQProducer() {

        TransactionMQProducer producer = new TransactionMQProducer(producerGroup);
        //nameserver地址
        producer.setNamesrvAddr(namesrvAddr);
        //事务回查最小并发数
        producer.setCheckThreadPoolMinSize(5);
        //事务回查最大并发数
        producer.setCheckThreadPoolMaxSize(20);
        //队列数
        producer.setCheckRequestHoldMax(2000);
        //服务器回调Producer，检查本地事务分支成功还是失败,虽然没用,但是还得设置
        producer.setTransactionCheckListener(new TransactionCheckListener() {
            @Override
            public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
                System.out.println("state -- " + new String(msg.getBody()));
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        return producer;
    }


}
